package com.hu.flink12.api.sql;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Author: hujianjun
 * @Date: 2021/2/9 21:55
 * @Describe: 使用SQL和Table做wordCount
 */
public class SqlOrTableForWordCount {
    public static void main(String[] args) throws Exception {
        // TODO 1.获取env和tableEnv
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings envSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, envSettings);


        // TODO 3.将DataStream数据转为Table或View,然后查询
        DataStream<String> lines = env.fromElements("hdfs hadoop hive", "hive hadoop", "hive flink");

        DataStream<String> words = lines.flatMap((String line, Collector<String> out) -> Arrays.stream(line.split(" ")).forEach(out::collect)).returns(Types.STRING);
        DataStream<Tuple2<String, Integer>> wordDataStream = words.map(word -> Tuple2.of(word, 1)).returns(Types.TUPLE(Types.STRING, Types.INT));

        // 3.1 使用Sql查询
//        tableEnv.createTemporaryView("wc_table", wordDataStream, $("word"), $("num"));
//        Table queryResult = tableEnv.sqlQuery("select word,sum(num) cnt from wc_table group by word");

        // 3.2 使用Table API查询
        Table wcTable = tableEnv.fromDataStream(wordDataStream, $("word"), $("num"));
        Table queryResult = wcTable.groupBy($("word"))
                .select($("word"), $("num").sum().as("cnt"))
                .filter($("word").isEqual("flink"));

        //打印schema信息
        queryResult.printSchema();

        //不能针对tableResult进行打印，需要转为DataStream才可以打印
//        DataStream<Tuple2<Boolean, Tuple2>> tuple2DataStream = tableEnv.toRetractStream(queryResult, Types.TUPLE(Types.STRING, Types.INT));
        DataStream<Tuple2<Boolean, Row>> tuple2DataStream = tableEnv.toRetractStream(queryResult, Row.class);

        // TODO 4.sink
        tuple2DataStream.print();

        // TODO 5.执行
        env.execute();
    }
}
